package com.linjiahao.nettyGateway.custom;

import com.linjiahao.commonall.constant.AuthConstant;
import com.linjiahao.nettyGateway.cluster.ClusterManager;
import com.linjiahao.nettyGateway.cluster.UserChannelInfo;
import com.linjiahao.nettyGateway.util.UserAgentUtils;
import io.netty.channel.Channel;
import io.netty.channel.ChannelFutureListener;
import org.apache.commons.logging.Log;
import org.apache.commons.logging.LogFactory;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.http.HttpHeaders;
import org.springframework.lang.Nullable;
import org.springframework.util.Assert;
import org.springframework.util.StringUtils;
import org.springframework.web.reactive.socket.HandshakeInfo;
import org.springframework.web.reactive.socket.WebSocketHandler;
import org.springframework.web.reactive.socket.WebSocketSession;
import org.springframework.web.reactive.socket.adapter.ReactorNettyWebSocketSession;
import org.springframework.web.reactive.socket.client.ReactorNettyWebSocketClient;
import reactor.core.publisher.Mono;
import reactor.netty.channel.ChannelOperations;
import reactor.netty.http.client.HttpClient;
import reactor.netty.http.websocket.WebsocketInbound;
import reactor.netty.http.websocket.WebsocketOutbound;

import java.net.InetSocketAddress;
import java.net.URI;
import java.util.Objects;

/**
*
* 自定义websocket连接，管理长连接
* */
public class CustomReactorNettyWebSocketClient extends ReactorNettyWebSocketClient {

    private static final Log logger = LogFactory.getLog(CustomReactorNettyWebSocketClient.class);

    private final HttpClient httpClient;


    @Nullable
    private Integer maxFramePayloadLength;


    private ClusterManager clusterManager;

    public CustomReactorNettyWebSocketClient() {
        this(HttpClient.create());
    }

    public CustomReactorNettyWebSocketClient(HttpClient httpClient) {
        this.maxFramePayloadLength = 65536;
        Assert.notNull(httpClient, "HttpClient is required");
        this.httpClient = httpClient;
    }

    public CustomReactorNettyWebSocketClient(
            HttpClient httpClient, ClusterManager clusterManager) {
        Assert.notNull(httpClient, "HttpClient is required");
        this.httpClient = httpClient;
        this.clusterManager = clusterManager;
    }

    @Override
    public HttpClient getHttpClient() {
        return this.httpClient;
    }


    @Deprecated
    @Override
    public void setMaxFramePayloadLength(int maxFramePayloadLength) {
        this.maxFramePayloadLength = maxFramePayloadLength;
    }


    @Deprecated
    @Override
    public int getMaxFramePayloadLength() {
        return this.maxFramePayloadLength;
    }



    @Override
    public Mono<Void> execute(URI url, WebSocketHandler handler) {
        return execute(url, new HttpHeaders(), handler);
    }

    @Override
    public Mono<Void> execute(URI url, HttpHeaders requestHeaders, WebSocketHandler handler) {
        String protocols = StringUtils.collectionToCommaDelimitedString(handler.getSubProtocols());
        return ((HttpClient.WebsocketSender)this.getHttpClient().headers((nettyHeaders) -> {
            this.setNettyHeaders(requestHeaders, nettyHeaders);
        }).websocket(protocols, this.getMaxFramePayloadLength()).uri(url.toString())).handle((inbound, outbound) -> {
            HttpHeaders responseHeaders = this.toHttpHeaders(inbound);
            String protocol = responseHeaders.getFirst("Sec-WebSocket-Protocol");
            HandshakeInfo info = new HandshakeInfo(url, responseHeaders, Mono.empty(), protocol);
            NettyDataBufferFactory factory = new NettyDataBufferFactory(outbound.alloc());
            WebSocketSession session = new ReactorNettyWebSocketSession(inbound, outbound, info, factory, this.getMaxFramePayloadLength());
            logger.info("WS Started session '" + session.getId() + "' for " + url);
            //添加网关通道
            String userId = requestHeaders.getFirst(AuthConstant.AUTH_HEADER);
            String clientType = requestHeaders.getFirst(AuthConstant.USER_CLIENT_TYPE_HEADER);
            addChannel(outbound, userId,clientType);
            return handler.handle(session).checkpoint(url + " [ReactorNettyWebSocketClient]");
        }).doOnRequest((n) -> {
            logger.info("Connecting to " + url);
        }).next();
    }

    private void setNettyHeaders(HttpHeaders httpHeaders, io.netty.handler.codec.http.HttpHeaders nettyHeaders) {
        httpHeaders.forEach(nettyHeaders::set);
    }

    private HttpHeaders toHttpHeaders(WebsocketInbound inbound) {
        HttpHeaders headers = new HttpHeaders();
        io.netty.handler.codec.http.HttpHeaders nettyHeaders = inbound.headers();
        nettyHeaders.forEach(entry -> {
            String name = entry.getKey();
            headers.put(name, nettyHeaders.getAll(name));
        });
        return headers;
    }



    /**
     * 添加通道
     *
     * @param websocketOutbound
     * @param uerId
     */
    private void addChannel(WebsocketOutbound websocketOutbound, String uerId,String clientType) {
        ChannelOperations channelOperations = (ChannelOperations) websocketOutbound;
        if (Objects.nonNull(clusterManager) && !StringUtils.isEmpty(uerId)) {
            Channel channel = channelOperations.channel();
            InetSocketAddress inetSocketAddress = (InetSocketAddress) channel.remoteAddress();
            UserChannelInfo userChannelInfo = UserChannelInfo.builder().userId(uerId).serverHost(inetSocketAddress.getHostString())
                    .channelId(channel.id().asLongText()).connectSystem(UserAgentUtils.getUserClient(clientType))
                    .serverPort(inetSocketAddress.getPort()).build();
            try {
                clusterManager.addChannel(channel,userChannelInfo);
                //channel关闭监听，自定义处理
                channel.closeFuture().addListener((ChannelFutureListener) future -> clusterManager.channelCloseHandle(userChannelInfo));
            }catch (Exception e){
                logger.info("gateway用户长连接管理异常:{}",e);
            }

        }
    }
}
